package com.lagou.handler;

import com.alibaba.fastjson.JSON;
import com.lagou.encoder.ResponseData;
import com.lagou.proto.RpcRequest;
import com.lagou.proto.RpcResponse;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class ResponseHandler extends ChannelInboundHandlerAdapter {

    private Map<String, ResponseFuture> responseFutureMap = new ConcurrentHashMap<String, ResponseFuture>(256);

    private ChannelHandlerContext channelHandlerContext;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ResponseData responseData = (ResponseData) msg;

        ResponseFuture responseFuture = responseFutureMap.get(responseData.getRequestId());
        if (responseFuture != null) {
            RpcResponse rpcResponse = new RpcResponse();
            rpcResponse.setResult(JSON.parseObject(responseData.getResultData(), responseFuture.getResultClass()));
            rpcResponse.setRequestId(responseData.getRequestId());
            responseFuture.putResponse(rpcResponse);
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        this.channelHandlerContext = ctx;
    }

    public RpcResponse invokeServer(final RpcRequest rpcRequest, Class resultClass) throws InterruptedException {
        final ResponseFuture responseFuture = new ResponseFuture(rpcRequest.getRequestId(), 20000, resultClass);
        this.responseFutureMap.put(rpcRequest.getRequestId(), responseFuture);
        channelHandlerContext.channel().writeAndFlush(rpcRequest).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture f) throws Exception {
                if (f.isSuccess()) {
                    responseFuture.setSendRequestOK(true);
                    return;
                } else {
                    responseFuture.setSendRequestOK(false);
                }

                responseFutureMap.remove(rpcRequest.getRequestId());
                responseFuture.setCause(f.cause());
                responseFuture.putResponse(null);
            }
        });
        RpcResponse rpcResponse = responseFuture.waitResponse(2000);
        if (rpcResponse == null) {
            if (responseFuture.isSendRequestOK()) {
                throw new RuntimeException("readTime out");
            } else {
                throw new RuntimeException("请求服务端失败", responseFuture.getCause());
            }
        }
        return rpcResponse;
    }

}
